package com.fanxuankai.boot.canal.listener.autoconfigure;

import com.fanxuankai.boot.canal.listener.MessageListener;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;

import java.util.Optional;

/**
 * @author fanxuankai
 */
@ConditionalOnClass({KafkaTemplate.class})
@ConditionalOnMissingClass({"com.fanxuankai.boot.mqbroker.producer.EventPublisher"})
@ConditionalOnProperty(name = CanalListenerProperties.ENABLED, matchIfMissing = true)
public class CanalListenerKafkaAutoConfiguration {
    @Bean
    public AbstractMessageListenerContainer<String, String> messageListenerContainer(
            ConsumerFactory<String, Object> consumerFactory, MessageListener messageListener,
            KafkaProperties kafkaProperties) {
        String[] topics = messageListener.getTopics().toArray(new String[0]);
        ContainerProperties containerProperties = new ContainerProperties(topics);
        containerProperties.setMessageListener((org.springframework.kafka.listener.MessageListener<String, String>) data ->
                messageListener.onMessage(data.topic(), data.value()));
        ConcurrentMessageListenerContainer<String, String> concurrentMessageListenerContainer =
                new ConcurrentMessageListenerContainer<>(consumerFactory, containerProperties);
        Optional.ofNullable(kafkaProperties.getListener().getConcurrency())
                .ifPresent(concurrentMessageListenerContainer::setConcurrency);
        containerProperties.getKafkaConsumerProperties().putAll(kafkaProperties.getConsumer().getProperties());
        return concurrentMessageListenerContainer;
    }
}
